/////////////////////////////////////////////////////////////////////////
//
// Amuse Engine SDK - system/ccm
// Copyright( c) 2014.  All Rights Reserved
//
// File:		AEChangeControlMgr.cpp
// Author:		Gianluca Belardelli
// Date:		25/06/2014
//
/////////////////////////////////////////////////////////////////////////
#include <AEEngine.h>

AE_THREAD_LOCAL(AEChangeControlMgr::NotifyList *) AEChangeControlMgr::m_tlsNotifyList;

AEChangeControlMgr::AEChangeControlMgr( void ) : m_lpTaskManager( 0L ), m_uiLastID( 0 ), m_csUpdate( 1000 )
{
	// Reserve some reasonable space to avoid delays because of multiple reallocations
	m_arrCumulativeNotifyList.Reserve( 4096 );
	m_arrSubjectsList.Reserve( 8192 );
	m_arrSubjectsList.SetSize( 1 );

    // Get ready to process changes in the main (this) thread
    NotifyList *lpList = new NotifyList();
    lpList->Reserve( 8192 );
	AE_THREAD_LOCAL_SET( m_tlsNotifyList, lpList );
	m_lstNotifyLists.AddElement( lpList );
}

AEChangeControlMgr::~AEChangeControlMgr( void )
{
	// Loop through all the subjects and their observers and clean up 
	for( AEUINT32 i=0; i<m_arrSubjectsList.GetSize(); i++)
	{
		if ( m_arrSubjectsList[i].m_lpSubject )
		{
			CCMObserversList &refObserversList = m_arrSubjectsList[i].m_arrObserversList;

			for( AEUINT32 k=0; k<refObserversList.GetSize(); k++ )
				Unregister( m_arrSubjectsList[i].m_lpSubject, refObserversList[k].m_lpObserver );
		}
	}

	m_lstNotifyLists.RemoveAll();
}

//////////////////////////////////////////////////////////////
// Implementazione dei metodi statici
//////////////////////////////////////////////////////////////
void AEChangeControlMgr::_initThreadLocalData( void *lpArg )
{
	AEASSERT_MSG( lpArg, "AEChangeControlMgr: No manager pointer passed to InitThreadLocalNotifyList" );

    AEChangeControlMgr *lpMgr = (AEChangeControlMgr *)lpArg;

	// Check if we have allocated a NotifyList for this thread.
	// The notify list is keep in tls (thread local storage).
	if( AE_THREAD_LOCAL_GET( lpMgr->m_tlsNotifyList ) == 0L )
	{
		NotifyList *lpList = new NotifyList();

		// Reserve some reasonable space to avoid delays because of multiple reallocations
		lpList->Reserve( 8192 );

		AE_THREAD_LOCAL_SET( lpMgr->m_tlsNotifyList, lpList );

		// Lock out the updates and add this NotifyList to m_NotifyLists
		AEScopedLock lock( lpMgr->m_csUpdate );
		lpMgr->m_lstNotifyLists.AddElement( lpList );
	}
}

void AEChangeControlMgr::_freeThreadLocalData( void *lpArg )
{
	AEASSERT_MSG( lpArg, "ChangeManager: No manager pointer passed to _freeThreadLocalData" );
    
	AEChangeControlMgr *lpMgr = (AEChangeControlMgr *)lpArg;

	// Free NotifyList if it exists
	delete static_cast<NotifyList*>( AE_THREAD_LOCAL_GET( lpMgr->m_tlsNotifyList ) );
	AE_THREAD_LOCAL_SET( lpMgr->m_tlsNotifyList, 0L );
}

void AEChangeControlMgr::_distributionCallback( void *lpParam, AEUINT32 uiBegin, AEUINT32 uiEnd )
{
	// Process the given range (this will be called from multiple threads)
	AEChangeControlMgr *lpMgr = static_cast<AEChangeControlMgr*>( lpParam );
	lpMgr->_distributeRange( uiBegin, uiEnd );
}

//////////////////////////////////////////////////////////////
// Implementazione dei metodi di AEChangeManagerI
//////////////////////////////////////////////////////////////
AEUINT32 AEChangeControlMgr::Register( AESubjectI *lpInSubject, AEUINT32 uiInIntrestBits, AEObserverI *lpInObserver, AEUINT32 uiObserverIdMask )
{
	AEUINT32 uiError = 0;

	if( lpInSubject && lpInObserver )
	{
		// Lock out updates while we register a subjext
		AEScopedLock lock( m_csUpdate );

		AEUINT32 uID = lpInSubject->GetID( this );

		if( uID != AESubject::AEInvalidSubjectID )
		{
			// Subject has already been registered. Add new observer to the list
			AESubjectInfo &si = m_arrSubjectsList[uID];
			
			si.m_arrObserversList.PushBack( AEObserverRequest( lpInObserver, uiInIntrestBits, uiObserverIdMask ) );
			uiInIntrestBits &= ~si.m_uiInterestBits;
			
			if( uiInIntrestBits )
			{
				si.m_uiInterestBits |= uiInIntrestBits;
				lpInSubject->UpdateInterestBits( this, uiInIntrestBits );
			}
		}
		else 
		{ 
			// New subject
			if ( m_arrFreeIDsList.IsEmpty() )
			{
				// No zero ID should ever be assigned, so use pre-increment
				uID = ++m_uiLastID;

				AEASSERT( uID == m_arrSubjectsList.GetSize() );
				
				m_arrSubjectsList.SetSize( uID + 1 );
			}
			else
			{
				uID = m_arrFreeIDsList.Back();
				m_arrFreeIDsList.PopBack();
			}

			AESubjectInfo &si = m_arrSubjectsList[uID];
			si.m_lpSubject = lpInSubject;
			si.m_arrObserversList.PushBack( AEObserverRequest( lpInObserver, uiInIntrestBits, uiObserverIdMask ) );
			si.m_uiInterestBits = uiInIntrestBits;

			lpInSubject->Attach( this, uiInIntrestBits, uID );
		}
		
		uiError = 1;
	}
	
	return uiError;
}

AEUINT32 AEChangeControlMgr::Unregister( AESubjectI *lpInSubject, AEObserverI *lpInObserver )
{
    AEUINT32 uiError = 1;

    if( lpInSubject && lpInObserver )
    {
        AEScopedLock lock( m_csUpdate );

        AEUINT32 uID = lpInSubject->GetID(this);
        if ( m_arrSubjectsList.GetSize() <= uID  ||  m_arrSubjectsList[uID].m_lpSubject != lpInSubject )
            return 1;

		AEASSERT( m_arrSubjectsList[uID].m_lpSubject == lpInSubject );

        CCMObserversList &observersList = m_arrSubjectsList[uID].m_arrObserversList;

		for( AEUINT32 i=0; i<observersList.GetSize(); i++)
		{
			if( observersList[i].m_lpObserver == lpInObserver )
			{
				observersList.RemoveAt(i);
				if( observersList.IsEmpty() )
				{
					m_arrSubjectsList[uID].m_lpSubject = 0L;
					m_arrFreeIDsList.PushBack( uID );
					lpInSubject->Detach( this );
				}
				
				uiError = 0;

				break;
			}
		}
    } 

    return uiError;
}

AEUINT32 AEChangeControlMgr::DistributeQueuedChanges( AEUINT32 uiSystemsToBeNotified, AEUINT32 uiChangesToDist )
{
	// Store the parameters so they can be used by multiple threads later
	m_uiSystemsToBeNotified = uiSystemsToBeNotified;
	m_uiChangesToDist = uiChangesToDist;

	// Loop through all the notifications.  We might need to loop through multiple
	// times because processing notifications might generate more notifications
	for(;;)
	{
		// Make sure m_indexList is big enough to hold all subjects
		m_arrIndexList.SetSize( m_arrSubjectsList.GetSize() );

		// Loop through all list and build m_cumulativeNotifyList
		for( AEUINT32 i=0; i<m_lstNotifyLists.GetSize(); i++ )
		{
			NotifyList *lpList = m_lstNotifyLists[i];
			size_t nOrigSize = lpList->GetSize();
			
			for( size_t i = 0; i < nOrigSize; i++ )
			{
				// Get notification
				AENotification &refNotif = lpList->operator[](i);

				// Get subject for notification
				AEUINT32 uID = refNotif.m_lpSubject->GetID( this );
				AEASSERT( uID != AESubject::AEInvalidSubjectID );

				if( uID != AESubject::AEInvalidSubjectID )
				{
					// Get the index for this subject
					AEUINT32 uiIndex = m_arrIndexList[uID];

					// If index is set, then this subject is already part of the m_cumulativeNotifyList
					if( uiIndex )
					{
						// Each subject only needs to be notified once for all changes
						// so let's combine all notifications for this subject
						m_arrCumulativeNotifyList[ uiIndex ].m_uiChangedBits |= refNotif.m_uiChangedBits;
					}
					else
					{
						// Set the index for this subject
						m_arrIndexList[uID] = (AEUINT32)m_arrCumulativeNotifyList.GetSize();

						// Add a new entry to m_cumulativeNotifyList
						m_arrCumulativeNotifyList.PushBack( AEMappedNotification( uID, refNotif.m_uiChangedBits ) );
					}
				}
			}

			// Clear out this list
			lpList->Clear();
		}

		// Determine number of notifications to process
		AEUINT32 uiNumberOfChanges = m_arrCumulativeNotifyList.GetSize();

		// Exit the loop if we don't have any messages to process
		if( uiNumberOfChanges == 0 )
			break;

		// If we have a task manager and there are more than 50 notifications to process, let's do it parallel
		static const AEUINT32 uiGrainSize = 50;

		if( m_lpTaskManager != NULL && (AEUINT32)uiNumberOfChanges > uiGrainSize )
		{
			// Process noticitions in parallel
			m_lpTaskManager->ParallelFor( NULL, _distributionCallback, this, 0, uiNumberOfChanges, uiGrainSize );
		}
		else
		{
			// Not enough notifications to worry about running in parallel, just distribute in this thread
			_distributeRange( 0, uiNumberOfChanges );
		}

		// Check if we are distributing all the notifications
		if( m_uiChangesToDist == 0xFFFFFFFF )
		{
			// If we are distributing all the notifications, clear out m_cumulativeNotifyList
			m_arrCumulativeNotifyList.Clear();
			m_arrIndexList.Clear();
		}
		else
		{
			// Some of the notifications might need to be distributed later
			// Keep m_cumulativeNotifyList and exit the loop
			break;
		}
	}

	return 0;
}

//////////////////////////////////////////////////////////////
// Implementazione dei metodi di AEObserverI
//////////////////////////////////////////////////////////////
AEUINT32 AEChangeControlMgr::ChangeOccurred( AESubjectI *lpSubject, AEUINT32 uiChangeTypeMask )
{
	AEUINT32 uiError = 0;

	AEASSERT( lpSubject );
	
	if( lpSubject )
	{
		if( !uiChangeTypeMask )
		{
			// The subject is shutting down - remove its reference
			uiError = _removeSubject( lpSubject );
		}
		else
		{
			// Get thread local notification list
			NotifyList &arrNotifyList = _getNotifyList();

			// IMPLEMENTATION NOTE
			// Don't check for duplicate instertions
			//
			// For the sake of performance and scalability don't do any operations 
			// that may require locks (requesting ID or checking shared data structures).
			// Frequent locking hurts incomparably more than even high percentage 
			// of duplicated insertions, especially taking into account that the memory 
			// is preallocated most of the time.
			arrNotifyList.PushBack( AENotification( lpSubject, uiChangeTypeMask ) );

			uiError = 1;
		}
	}

	return uiError;
}

//////////////////////////////////////////////////////////////
// Altri metodi pubblici
//////////////////////////////////////////////////////////////
AEUINT32 AEChangeControlMgr::SetTaskManager( AEITaskManager *lpTaskMgr )
{
	AEASSERT_MSG( lpTaskMgr, "AEChangeControlMgr::SetTaskManager lpTaskMgr invalido." );
	AEASSERT_MSG (!m_lpTaskManager, "ChangeManager: Call ResetTaskManager before using SetTaskManager to set the new task manager");

	// Store TaskManager
	m_lpTaskManager = lpTaskMgr;

	// Make each thread call InitThreadLocalData
	m_lpTaskManager->NonStandardPerThreadCallback( _initThreadLocalData, this );

	return 0;
}

void AEChangeControlMgr::ResetTaskManager( void )
{
	// Free all data associated with m_pTaskManager
	if ( m_lpTaskManager )
	{
		// Make each thread call FreeThreadLocalData
		m_lpTaskManager->NonStandardPerThreadCallback( _freeThreadLocalData, this );
		m_lstNotifyLists.RemoveAll();
		m_lpTaskManager = 0L;

		// Restore main (this) thread data
		NotifyList *lpList = new NotifyList();

		AE_THREAD_LOCAL_SET( m_tlsNotifyList, lpList );

		m_lstNotifyLists.AddElement( lpList );
	}
}

//////////////////////////////////////////////////////////////
// Implementazione dei metodi privati
//////////////////////////////////////////////////////////////
AEUINT32 AEChangeControlMgr::_removeSubject ( AESubjectI *lpSubject )
{
	AEUINT32 uiError = 0;

	CCMObserversList refObserversList;
	{
		AEScopedLock lock( m_csUpdate );

		AEUINT32 uID = lpSubject->GetID( this );
		AEASSERT( uID != AESubject::AEInvalidSubjectID );
		AEASSERT( m_arrSubjectsList[uID].m_lpSubject == lpSubject );

		if ( m_arrSubjectsList.GetSize() <= uID  ||  m_arrSubjectsList[uID].m_lpSubject != lpSubject )
			return 1;

		refObserversList = m_arrSubjectsList[uID].m_arrObserversList;
		m_arrSubjectsList[uID].m_lpSubject = 0L;
		m_arrFreeIDsList.PushBack( uID );
		uiError = 0;
	}

	for( AEUINT32 i=0; i < refObserversList.GetSize(); i++ )
		lpSubject->Detach( refObserversList[i].m_lpObserver );

	return uiError;
}

void AEChangeControlMgr::_distributeRange ( AEUINT32 uiBegin, AEUINT32 uiEnd )
{
	// Loop through all the noticatiosn in the given range
	for ( AEUINT32 i = uiBegin; i < uiEnd; ++i )
	{
		// Get the notification and the subject
		AEMappedNotification &refNotif = m_arrCumulativeNotifyList[ i ];
		AESubjectInfo &refSubject = m_arrSubjectsList[ refNotif.m_uiSubjectID ];

		// Distribute any desired changes
		AEUINT32 uiActiveChanges = refNotif.m_uiChangedBits & m_uiChangesToDist;
		if( uiActiveChanges )
		{
			// Clear the bit for the changes we are distributing
			refNotif.m_uiChangedBits &= ~uiActiveChanges;

			// Loop through all the observers and let them process the notifictaion
			CCMObserversList &obsList = refSubject.m_arrObserversList;
			for( AEUINT32 k=0; k<obsList.GetSize(); k++ )
			{
				// Determine if this observe is interested in this notification
				AEUINT32 uiChangesToSend = obsList[k].m_uiInterestBits & uiActiveChanges;
				if ( uiChangesToSend )
				{
					// If this observer is part of the systems to be notified then we can pass it this notification
					if( obsList[k].m_uiObserverIdBits & m_uiSystemsToBeNotified )
					{
						// Have the observer process this change (notification)
						obsList[k].m_lpObserver->ChangeOccurred( refSubject.m_lpSubject, uiChangesToSend );
					}
				}
			}
		}
	}
}

AE_FORCEINLINE AEChangeControlMgr::NotifyList &AEChangeControlMgr::_getNotifyList( void )
{
	return *static_cast<AEChangeControlMgr::NotifyList*>( AE_THREAD_LOCAL_GET( m_tlsNotifyList ) );
}
